package com.bclz.cat.uniquekey;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import java.io.IOException;
import java.util.*;

/**
 * 对每个Map进行排序，获得topN,减少带宽以及资源损害
 */
public class CatMapper extends Mapper<LongWritable,Text,NullWritable,Text> {

    //default 10
    private  int N=10;
    //同一个map中的topN
    private SortedMap<Integer,Text> tree=new TreeMap<>((a,b)->b-a);

    @Override
    protected void setup(Context context) throws IOException, InterruptedException {

        Configuration configuration = context.getConfiguration();
        Integer topN = configuration.getInt("hadoop.top.n",10);
        this.N=topN;
    }

    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {

        String lineStr=value.toString();
        String[] split = lineStr.split(",");

        if(split.length>2){

            Integer count=Integer.valueOf(split[2]);

            //直接set会导致对象引用，所有的tree value都为最后一次的输入value
            tree.put(count,new Text(lineStr));
        }


        if(tree.size()>N){
            tree.remove(tree.lastKey());
        }
    }

    /**
     * 单个map执行完后写入context
     * @param context
     * @throws IOException
     * @throws InterruptedException
     */
    @Override
    protected void cleanup(Context context) throws IOException, InterruptedException {
        for (Text v:
             tree.values()) {
            context.write(NullWritable.get(),v);
        }


    }


}
